package com.example.flink;

import com.example.entity.CreditBehavior;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;

/**
 * 信贷行为Flink任务 - 暂时禁用
 */
@Slf4j
// @Component
public class CreditBehaviorJob {

    @Autowired
    private StreamExecutionEnvironment env;

    @Value("${flink.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${flink.kafka.group-id}")
    private String groupId;

    @Value("${flink.kafka.topics.credit-data}")
    private String creditDataTopic;

    /**
     * 静态内部类 - 信贷行为数据映射器
     */
    public static class CreditBehaviorMapper implements MapFunction<String, CreditBehavior> {
        private static final ObjectMapper objectMapper = new ObjectMapper();

        @Override
        public CreditBehavior map(String value) throws Exception {
            try {
                JsonNode jsonNode = objectMapper.readTree(value);
                CreditBehavior behavior = new CreditBehavior();

                // 设置基本字段
                behavior.setUserId(jsonNode.path("userId").asLong());
                behavior.setCurrentLoanCount(jsonNode.path("currentLoanCount").asInt(0));
                behavior.setCurrentCreditCardCount(jsonNode.path("currentCreditCardCount").asInt(0));
                behavior.setHistoricalOverdueCount(jsonNode.path("historicalOverdueCount").asInt(0));
                behavior.setCurrentOverdueAmount(new BigDecimal(jsonNode.path("currentOverdueAmount").asText("0")));
                behavior.setM1OverdueCount(jsonNode.path("m1OverdueCount").asInt(0));
                behavior.setM2OverdueCount(jsonNode.path("m2OverdueCount").asInt(0));
                behavior.setM3OverdueCount(jsonNode.path("m3OverdueCount").asInt(0));

                // 处理时间字段
                String lastLoanTimeStr = jsonNode.path("lastLoanTime").asText(null);
                String lastRepaymentTimeStr = jsonNode.path("lastRepaymentTime").asText(null);
                if (lastLoanTimeStr != null && !lastLoanTimeStr.isEmpty()) {
                    behavior.setLastLoanTime(LocalDateTime.parse(lastLoanTimeStr));
                }
                if (lastRepaymentTimeStr != null && !lastRepaymentTimeStr.isEmpty()) {
                    behavior.setLastRepaymentTime(LocalDateTime.parse(lastRepaymentTimeStr));
                }

                behavior.setCreditInquiryLast1Month(jsonNode.path("creditInquiryLast1Month").asInt(0));
                behavior.setCreditInquiryLast3Months(jsonNode.path("creditInquiryLast3Months").asInt(0));
                behavior.setCreditInquiryLast6Months(jsonNode.path("creditInquiryLast6Months").asInt(0));
                behavior.setTotalCreditLine(new BigDecimal(jsonNode.path("totalCreditLine").asText("0")));
                behavior.setUsedCreditLine(new BigDecimal(jsonNode.path("usedCreditLine").asText("0")));
                behavior.setPbocCreditScore(jsonNode.path("pbocCreditScore").asInt(0));
                behavior.setBaihangCreditScore(jsonNode.path("baihangCreditScore").asInt(0));
                behavior.setSesameScore(jsonNode.path("sesameScore").asInt(0));
                behavior.setCreateTime(LocalDateTime.now());
                behavior.setUpdateTime(LocalDateTime.now());

                return behavior;
            } catch (Exception e) {
                log.error("解析信贷数据失败: {}", value, e);
                return null;
            }
        }
    }

    /**
     * 静态内部类 - 信贷行为处理器
     */
    public static class CreditBehaviorProcessor extends ProcessWindowFunction<CreditBehavior, CreditBehavior, Long, TimeWindow> {
        @Override
        public void process(Long userId, ProcessWindowFunction<CreditBehavior, CreditBehavior, Long, TimeWindow>.Context context,
                           Iterable<CreditBehavior> elements, Collector<CreditBehavior> out) {
            List<CreditBehavior> behaviors = new ArrayList<>();
            elements.forEach(behaviors::add);

            if (behaviors.isEmpty()) {
                return;
            }

            // 获取最新的信贷行为数据
            CreditBehavior latestBehavior = behaviors.get(behaviors.size() - 1);

            // 计算信用卡使用率
            if (latestBehavior.getTotalCreditLine().compareTo(BigDecimal.ZERO) > 0) {
                BigDecimal utilizationRate = latestBehavior.getUsedCreditLine()
                        .divide(latestBehavior.getTotalCreditLine(), 4, RoundingMode.HALF_UP);
                latestBehavior.setCreditCardUtilizationRate(utilizationRate);
            }

            // 输出处理后的信贷行为数据
            out.collect(latestBehavior);

            log.info("处理用户[{}]的信贷行为数据完成", userId);
        }
    }

    public void init() {
        try {
            // 创建Kafka数据源
            KafkaSource<String> source = KafkaSource.<String>builder()
                    .setBootstrapServers(bootstrapServers)
                    .setTopics(creditDataTopic)
                    .setGroupId(groupId)
                    .setStartingOffsets(OffsetsInitializer.latest())
                    .setValueOnlyDeserializer(new SimpleStringSchema())
                    .build();

            // 创建数据流
            DataStream<String> stream = env.fromSource(
                    source,
                    WatermarkStrategy.noWatermarks(),
                    "Kafka Source");

            // 解析JSON数据并转换为CreditBehavior对象
            DataStream<CreditBehavior> creditStream = stream
                    .map(new CreditBehaviorMapper())
                    .filter(behavior -> behavior != null);

            // 按用户ID分组，计算信贷行为指标
            creditStream
                    .keyBy(CreditBehavior::getUserId)
                    .window(TumblingProcessingTimeWindows.of(Time.minutes(5)))
                    .process(new CreditBehaviorProcessor());

            // 启动Flink任务
            env.execute("Credit Behavior Job");
        } catch (Exception e) {
            log.error("启动信贷行为Flink任务失败", e);
        }
    }
}